package com.tang.transform;

import com.tang.bean.WaterSensor;
import com.tang.functions.MapFunctionImpl;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * map
 *
 * @author tang
 * @since 2023/6/6 22:34
 */
public class MapDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<WaterSensor> source = env.fromElements(
                new WaterSensor("s1", 1L, 1),
                new WaterSensor("x2", 2L, 2),
                new WaterSensor("r3", 3L, 3)
        );

        // 方式一：匿名内部类
        /*source.map(new MapFunction<WaterSensor, String>() {
            @Override
            public String map(WaterSensor waterSensor) throws Exception {
                return waterSensor.id;
            }
        }).returns(Types.STRING()).print();*/

        // 方式二：lambda表达式
        //source.map(sensor -> sensor.id).print();
        //source.map(WaterSensor::getId).print();

        // 方式三：一个实现类MapFunction的实现类
        //source.map(new MyMapFunction()).print();
        source.map(new MapFunctionImpl()).print();

        env.execute();
    }

    public static class MyMapFunction implements MapFunction<WaterSensor, String> {

        @Override
        public String map(WaterSensor waterSensor) throws Exception {
            return waterSensor.id;
        }

    }

}
